package small.common.statistic.base;

import small.common.core.util.ParamUtil;
import small.common.core.util.TimeUtil;
import small.common.statistic.data.WindowWrap;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 存储统计数据
 * <p>
 * 基于时间轮（Time Wheel）算法的实现，用于存储并统计一个固定时间窗口内的度量数据。
 * 它由多个Bucket组成，每个Bucket代表一个时间段，例如1秒或1分钟。
 * LeapArray支持并发写入和读取操作，并可以通过快速定位到当前时间所在的Bucket来实现高效的数据更新和清理。
 * 场景：用于流量控制、熔断降级等场景中，帮助用户对系统进行实时监控和调整。
 *
 * @author jinzhou
 * @data 2023/4/7 11:13
 */
public abstract class LeapArray<T> {

    /**
     * 窗口长度（单位：Ms）
     */
    protected int windowLengthInMs;

    /**
     * 样本计数
     */
    protected int sampleCount;

    /**
     * 间隔（单位：Ms）
     */
    protected int intervalInMs;

    /**
     * 间隔（秒）使用 可以包含小数值，以提供更精确的时间间隔
     */
    private double intervalInSecond;

    /**
     * 窗口
     */
    protected final AtomicReferenceArray<WindowWrap<T>> array;

    private final ReentrantLock updateLock = new ReentrantLock();

    public LeapArray(int sampleCount, int intervalInMs) {
        ParamUtil.isTrue(sampleCount > 0, "bucket count is invalid: " + sampleCount);
        ParamUtil.isTrue(intervalInMs > 0, "total time interval of the sliding window should be positive");
        ParamUtil.isTrue(intervalInMs % sampleCount == 0, "time span needs to be evenly divided");

        this.windowLengthInMs = intervalInMs / sampleCount;
        this.intervalInMs = intervalInMs;
        this.intervalInSecond = intervalInMs / 1000.0;
        this.sampleCount = sampleCount;
        this.array = new AtomicReferenceArray<>(sampleCount);
    }


    /**
     * 获取当前窗口
     *
     * @return
     */
    public WindowWrap<T> currentWindow() {
        return currentWindow(TimeUtil.currentTimeMillis());
    }

    /**
     * 获取上个窗口
     *
     * @return
     */
    public WindowWrap<T> getPreviousWindow() {
        return getPreviousWindow(TimeUtil.currentTimeMillis());
    }

    /**
     * 获取时间戳所在窗口
     *
     * @param timeMillis
     * @return
     */
    public WindowWrap<T> currentWindow(long timeMillis) {
        if (timeMillis < 0) {
            return null;
        }
        // 获取当前时间在窗口数组中映射的下标
        int idx = calculateTimeIdx(timeMillis);
        // 计算当前时间对应的窗口的开始时间
        long windowStart = calculateWindowStart(timeMillis);
        while (true) {
            WindowWrap<T> old = array.get(idx);
            /*
             * 根据下标在环形数组中获取滑动窗口（桶）
             *
             * (1) 如果桶不存在则创建新的桶，并通过CAS将新桶赋值到数组下标位。
             * (2) 如果获取到的桶不为空，并且桶的开始时间等于刚刚算出来的时间，那么返回当前获取到的桶。
             * (3) 如果获取到的桶不为空，并且桶的开始时间小于刚刚算出来的开始时间，那么说明这个桶是上一圈用过的桶，重置当前桶
             * (4) 如果获取到的桶不为空，并且桶的开始时间大于刚刚算出来的开始时间，理论上不应该出现这种情况。
             */
            if (old == null) {
                // 第一次进入，新建窗口，并使用cas的方式设置，如果出现争抢导致设置失败，暂时让出执行权待其它线程成功设置
                WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
                if (array.compareAndSet(idx, null, window)) {
                    // 更新成功，返回创建的bucket
                    return window;
                } else {
                    // 争用失败，线程将返回其时间片以等待bucket可用。
                    Thread.yield();
                }
            } else if (windowStart == old.windowStart()) {
                // 当前时间对应的窗口开始时间等于获取到的窗口开始时间，那么当前获取到的窗口就是我们需要的
                return old;
            } else if (windowStart > old.windowStart()) {
                // 当前时间对应的窗口开始时间大于获取到的窗口开始时间，那么当前获取到的窗口为已过期窗口，加锁重置
                if (updateLock.tryLock()) {
                    try {
                        // 重置当前窗口
                        return resetWindowTo(old, windowStart);
                    } finally {
                        updateLock.unlock();
                    }
                } else {
                    //放弃当前线程时间，重新获取
                    Thread.yield();
                }
            } else if (windowStart < old.windowStart()) {
                // 容错机制（发生了时间回溯）:返回空数据
                return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            }
        }
    }

    /**
     * 获取时间戳所在的上个窗口
     *
     * @param timeMillis
     * @return
     */
    public WindowWrap<T> getPreviousWindow(long timeMillis) {
        if (timeMillis < 0) {
            return null;
        }
        //上个窗口的下标
        int idx = calculateTimeIdx(timeMillis - windowLengthInMs);
        //上个窗口时间
        timeMillis = timeMillis - windowLengthInMs;
        WindowWrap<T> wrap = array.get(idx);

        //没有窗口 或者 是弃用窗口
        if (wrap == null || isWindowDeprecated(wrap)) {
            return null;
        }
        //超出当前窗口统计范围
        if (wrap.windowStart() + windowLengthInMs < (timeMillis)) {
            return null;
        }

        return wrap;
    }

    /**
     * 获取同一个窗口下标
     *
     * @param timeMillis
     * @return
     */
    private int calculateTimeIdx(long timeMillis) {
        // 利用除法取整原则，保证了一个窗口内的所有时间所得到的timeId是相等的
        long timeId = timeMillis / windowLengthInMs;
        // 利用求余运算原则，保证一个窗口内获取到的桶的下标位是一致的
        return (int) (timeId % this.array.length());
    }

    /**
     * 获取同一个窗口开始时间
     *
     * @param timeMillis
     * @return
     */
    private long calculateWindowStart(long timeMillis) {
        // 利用求余运算原则，保证一个窗口内获取到的桶的开始时间是一致的
        return timeMillis - timeMillis % windowLengthInMs;
    }

    /**
     * 获取当前时间对应的值
     *
     * @param timeMillis
     * @return
     */
    public T getWindowValue(long timeMillis) {
        if (timeMillis < 0) {
            return null;
        }
        int idx = calculateTimeIdx(timeMillis);

        WindowWrap<T> bucket = array.get(idx);

        if (bucket == null || !bucket.isTimeInWindow(timeMillis)) {
            return null;
        }

        return bucket.value();
    }

    public List<T> values() {
        return values(TimeUtil.currentTimeMillis());
    }

    public List<T> values(long timeMillis) {
        if (timeMillis < 0) {
            return new ArrayList<T>();
        }
        int size = array.length();
        List<T> result = new ArrayList<T>(size);

        for (int i = 0; i < size; i++) {
            WindowWrap<T> windowWrap = array.get(i);
            // 过滤掉过期数据
            if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
                continue;
            }
            result.add(windowWrap.value());
        }
        return result;
    }

    /**
     * 判断是否在窗口统计范围内  如果不在表示已弃用
     *
     * @param windowWrap
     * @return
     */
    public boolean isWindowDeprecated(WindowWrap<T> windowWrap) {
        return isWindowDeprecated(TimeUtil.currentTimeMillis(), windowWrap);
    }

    /**
     * 判断是否在窗口统计范围内 如果不在表示已弃用
     *
     * @param time
     * @param windowWrap
     * @return
     */
    public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
        return time - windowWrap.windowStart() > intervalInMs;
    }

    /**
     * 获取时间窗口间隔（秒）
     *
     * @return
     */
    public double getIntervalInSecond() {
        return intervalInSecond;
    }

    /**
     * 重置窗口的值
     *
     * @param windowWrap
     * @param startTime
     * @return
     */
    protected abstract WindowWrap<T> resetWindowTo(WindowWrap<T> windowWrap, long startTime);

    /**
     * 创建一个新窗口
     *
     * @param timeMillis
     * @return
     */
    public abstract T newEmptyBucket(long timeMillis);

}